package club.monkeywood.ad.dmp.report

import club.monkeywood.ad.dmp.bean.Log
import club.monkeywood.ad.dmp.report.AppAnalyseRpt.{broadcast, inputPath, outputPath, ss}
import club.monkeywood.ad.dmp.util.{JedisPools, RptUtils}
import org.apache.commons.lang.StringUtils
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession


/**
	* 媒体报表分析
	*     使用redis替代广播变量查询app id对应的app name
	*/
object AppAnalyseRpt2RedisByRDD extends App {

	if (args.length != 2) {
		println(
			"""
				|cn.dmp.report.AppAnalyseRptV2
				|参数：
				| 输入路径
				| 输出路径
			""".stripMargin)
		sys.exit()
	}

	// 2 创建sparkconf->SparkSession
	val sparkConf = new SparkConf()
	sparkConf.setAppName(s"${this.getClass.getSimpleName}")
	sparkConf.setMaster("local[*]")
	//使用KryoSerializer更快
	// RDD 序列化到磁盘 worker与worker之间的数据传输
	sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

	val ss = SparkSession.builder()
		.appName(s"${this.getClass.getSimpleName}")
		.config(sparkConf)
		.getOrCreate()

	// 读取数据
	ss.sparkContext.textFile(inputPath)
		.map(_.split(",", -1))
		.filter(_.length >=85)
		.map(Log(_)).filter(log => !log.appid.isEmpty || !log.appname.isEmpty)
  	.mapPartitions( itr => {
			  val jedis = JedisPools.getConnection()

			  val parResult = new collection.mutable.ListBuffer[(String, List[Double])]()

			  // 遍历分区的所有数据，查询redis（把appname为空的数据进行转换），将结果放入到Listbuffer
			  itr.foreach(log => {
				  var newAppName = log.appname
				  if (StringUtils.isEmpty(newAppName)) {
					  newAppName = jedis.get(log.appid)
				  }

				  val req = RptUtils.caculateReq(log.requestmode, log.processnode)
				  val rtb = RptUtils.caculateRtb(log.iseffective, log.isbilling, log.isbid, log.adorderid, log.iswin, log.winprice, log.adpayment)
				  val showClick = RptUtils.caculateShowClick(log.requestmode, log.iseffective)

				  parResult += ((newAppName, req ++ rtb ++ showClick))
			  })

			  jedis.close()

			  // 将listbuffer转成迭代器
		    // mapPartitions需要返回迭代器
			  parResult.toIterator
		  }
	  ).reduceByKey((list1, list2) => {
			list1.zip(list2).map(t => t._1 + t._2)
		}).map(t => t._1 + "," + t._2.mkString(","))
			.saveAsTextFile(outputPath)

	ss.stop()

}
